package com.kai.service.mqtt.handler;

import com.kai.service.mqtt.consumer.MessageConsumerAbstract;
import com.kai.service.mqtt.consumer.TestConsumer;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttMessage;

import java.nio.charset.StandardCharsets;
import java.util.Objects;

@Data
@Slf4j
public class MessageHandler {

    private String topic;

    private MqttMessage mqttMessage;

    private MessageConsumerAbstract messageConsumer;

    public MessageHandler(String topic, MqttMessage mqttMessage) {
        this.topic = topic;
        this.mqttMessage = mqttMessage;
        this.setConsumer();
    }

    public void handle() {
        if (Objects.nonNull(messageConsumer)) {
            messageConsumer.consume(mqttMessage);
        }
    }

    private void setConsumer() {
        String body = new String(mqttMessage.getPayload(), StandardCharsets.UTF_8);
        log.info("receive msg from topic:{}, body is:{} ", topic, body);

        if (topic.startsWith("TEST/")) {
            messageConsumer = new TestConsumer();
        }
    }
}
